package com.wechat.network.remote;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.swak.Constants;
import com.swak.OS;
import com.swak.exception.InvokeException;
import com.swak.exception.RemotingException;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.transport.ExchangeClient;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.Exchangers;
import com.swak.reactivex.transport.Message;
import com.swak.reactivex.transport.TransportMode;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.Ints;
import com.swak.utils.Maps;
import com.swak.utils.StringUtils;
import com.wechat.command.Command;
import com.wechat.command.CommandResult;
import com.wechat.network.AbstractNetwork;
import com.wechat.network.Network;
import com.wechat.network.NetworkHandler;
import com.wechat.network.NetworkHandler.NetworkExchangeHandler;

/**
 * 远端服务
 * 
 * @author 618lf
 */
public class RemoteNetwork extends AbstractNetwork implements Network {

	private static final String Riad = "BeeChat";

	private final AtomicBoolean closed = new AtomicBoolean(true);
	private final LoopResources resources;
	private final URL url;
	private final NetworkHandler handler;
	private ExchangeClient client;

	public RemoteNetwork(String serverAddrs, NetworkHandler handler) {
		this.resources = Contexts.createEventLoopResources(TransportMode.OS, 1, 4, Riad + ".", false, 2,
				TimeUnit.SECONDS);
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Bind_Host, Constants.ANYHOST);
		parameters.put(Constants.Heartbeat, String.valueOf(Constants.Default_Heartbeat));
		this.url = new URL(Riad, OS.ip(), 10889, parameters);
		this.handler = handler;
	}

	@Override
	public void connect() {
		if (!this.closed.get()) {
			return;
		}
		try {
			this.client = Exchangers.connect(url, resources, new ChatExchangeHandler(this.handler));
		} catch (Exception e) {
			this.closed.set(true);
			throw new IllegalArgumentException("启动异常.", e);
		}
	}

	@Override
	public CompletionStage<CommandResult> send(Command message) {

		if (logger.isDebugEnabled()) {
			logger.debug("[网络] 发送消息 {}", message.getData());
		}

		// 网络已经关闭
		if (this.closed.get()) {
			return CompletableFuture.completedFuture(CommandResult.fail(StringUtils.format("网络已经关闭！")));
		}

		// 通过 TCP 发送数据
		CompletableFuture<CommandResult> resultFuture = new CompletableFuture<>();
		this.retryExecuteRunnable(() -> {
			try {
				CompletableFuture<Object> future = this.client.request(message);
				future.whenComplete((r, e) -> {
					if (e != null) {
						resultFuture.completeExceptionally(e);
					} else if (r != null && r instanceof Message) {
						resultFuture.complete(CommandResult.wrap(((Message) r).getData()));
					} else {
						resultFuture.complete(CommandResult.wrap(r));
					}
				});
			} catch (RemotingException e) {
				throw new InvokeException("Send message error", e);
			}
		});
		return resultFuture;
	}

	private void retryExecuteRunnable(Runnable runnable) {
		this.retryExecuteRunnable(3, runnable);
	}

	private void retryExecuteRunnable(int times, Runnable runnable) {
		int tries = Ints.max(times, 3);
		while (tries > 0) {
			try {
				runnable.run();
				break;
			} catch (Exception e) {
			}
			tries--;
		}
	}

	/**
	 * Chat 消息处理
	 * 
	 * @author 618lf
	 */
	class ChatExchangeHandler extends NetworkExchangeHandler implements ExchangeHandler {

		ChatExchangeHandler(NetworkHandler delegate) {
			super(RemoteNetwork.this, delegate);
		}
	}
}